package com.simon.cdc.sink;

import com.simon.cdc.database.DatabaseConnectionConfig;
import com.simon.cdc.database.DatabaseConnectionPool;
import com.simon.cdc.event.ChangeEvent;
import com.simon.cdc.event.ChangeEventParser;
import com.simon.cdc.event.ChangeEventProcessor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

/**
 * 基于Druid连接池的MySQL CDC数据同步Sink
 */
public class MySQLDruidSink extends RichSinkFunction<String> {

    private final DatabaseConnectionConfig connectionConfig;
    private transient DatabaseConnectionPool connectionPool;
    private transient ChangeEventProcessor eventProcessor;

    public MySQLDruidSink(DatabaseConnectionConfig connectionConfig) {
        this.connectionConfig = connectionConfig;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        this.connectionPool = new DatabaseConnectionPool(connectionConfig);
        this.eventProcessor = new ChangeEventProcessor();
    }

    @Override
    public void invoke(String jsonEvent, Context context) throws Exception {
        ChangeEvent event = ChangeEventParser.parse(jsonEvent);
        eventProcessor.process(event, connectionPool);
    }

    @Override
    public void close() throws Exception {
        connectionPool.close();
        super.close();
    }
}